package com.zzl.kafkatest;

import kafka.consumer.*;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.serializer.Decoder;
import kafka.utils.VerifiableProperties;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class MyConsumer {

    private final ConsumerConnector consumer;
    private final static int  partitions =1;
    public MyConsumer(){
        Properties originalProps = new Properties();

        //zookeeper 配置，通过zk 可以负载均衡的获取broker
        originalProps.put("zookeeper.connect", "172.31.0.91:2182");

        //group 代表一个消费组
        originalProps.put("group.id", "zzl11");

        //zk连接超时时间
        originalProps.put("zookeeper.session.timeout.ms", "10000");
        //zk同步时间
        originalProps.put("zookeeper.sync.time.ms", "2000");
        //自动提交间隔时间
//        originalProps.put("auto.commit.interval.ms", "1000");
        originalProps.put("enable.auto.commit", "false");
        originalProps.put("max.poll.records", 200);
        //消息日志自动偏移量,防止宕机后数据无法读取
        originalProps.put("auto.offset.reset", "smallest");
        //序列化类
        originalProps.put("serializer.class", "kafka.serializer.StringEncoder");

        //构建consumer connection 对象
        consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(originalProps));
    }

    public void consume(){
        //指定需要订阅的topic
        Map<String ,Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(MyProducer.TOPIC, new Integer(partitions));

        //指定key的编码格式
        Decoder<String> keyDecoder = new kafka.serializer.StringDecoder(new VerifiableProperties());
        //指定value的编码格式
        Decoder<String> valueDecoder = new kafka.serializer.StringDecoder(new VerifiableProperties());

        //获取topic 和 接受到的stream 集合
        Map<String, List<KafkaStream<String, String>>> map = consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);

        //根据指定的topic 获取 stream 集合
        List<KafkaStream<String, String>> kafkaStreams = map.get(MyProducer.TOPIC);
        ExecutorService executor = Executors.newFixedThreadPool(partitions);

        //因为是多个 message组成 message set ， 所以要对stream 进行拆解遍历
        for(final KafkaStream<String, String> kafkaStream : kafkaStreams){

            executor.submit(new Runnable() {

                @Override
                public void run() {
                    //拆解每个的 stream
                    ConsumerIterator<String, String> iterator = kafkaStream.iterator();

                    while (iterator.hasNext()) {

                        //messageAndMetadata 包括了 message ， topic ， partition等metadata信息
                        MessageAndMetadata<String, String> messageAndMetadata = iterator.next();
                        messageAndMetadata.partition();
                        messageAndMetadata.offset();
                        String format = String.format("thread-id=%d\nreceive:key = %s\nvalue=%\noffset=%d\npartition=%d\n\n",Thread.currentThread().getId(), messageAndMetadata.key(), messageAndMetadata.message(), messageAndMetadata.offset(), messageAndMetadata.partition());
                        System.out.println(format);
//                        System.out.println( Thread.currentThread().getId() +"-message : " + messageAndMetadata.message() + "  partition :  " + messageAndMetadata.partition());

                    }
                }
            });

        }
        executor.shutdown();
    }

    public static void main(String[] args) {
        new MyConsumer().consume();
    }

}
